package com.hub.realtime.flink.submit.client;

import com.hub.realtime.common.exception.UtilException;
import com.hub.realtime.common.model.FlinkRunRequest;
import com.hub.realtime.common.model.FlinkStopRequest;
import com.hub.realtime.common.model.FlinkSubmitResponse;
import com.hub.realtime.flink.submit.base.FlinkSubmitBase;
import com.hub.realtime.flink.submit.base.FlinkYarnSubmitBase;
import com.hub.realtime.flink.submit.impl.FlinkYarnApplicationSubmit;
import lombok.SneakyThrows;

public class SubmitClient {
    /**
     * 提交任务
     *
     * @param flinkRunRequest
     * @return
     */
    public static FlinkSubmitResponse runSubmit(FlinkRunRequest flinkRunRequest) throws UtilException {
        FlinkSubmitBase flinkSubmit = null;
        if (flinkRunRequest.getExecuteMode().equals("yarn-application")) {
            flinkSubmit = new FlinkYarnApplicationSubmit();
            return flinkSubmit.submit(flinkRunRequest);
        } else {
            throw new UtilException("暂时不支持该模式提交");
        }

    }

    /**
     * 停止任务
     *
     * @param flinkStopRequest
     * @return
     */
    @SneakyThrows
    public static String runStop(FlinkStopRequest flinkStopRequest) {
        FlinkYarnSubmitBase flinkYarnSubmitBase = new FlinkYarnApplicationSubmit();
        return flinkYarnSubmitBase.stop(flinkStopRequest);
    }
}
